package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"flag"
	"fmt"
	"math/rand"
	"os"
	"runtime"
	"sort"
	"strings"
	"sync"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/rs/zerolog"
)

// Config holds the configuration for data generation
type Config struct {
	SourceDSN          string        // 源数据库连接字符串
	TargetDSN          string        // 目标数据库连接字符串
	SchemaName         string        // 数据库模式名
	TableTypesName     string        // 各种类型测试表名
	TableCompositeName string        // 复合键测试表名
	TableEmptyName     string        // 空表表名
	RowCount           int           // 主表 (types/composite) 要生成的总行数 (百万级)
	BatchSize          int           // 批量插入的行数
	ModifyRatio        float64       // 目标库中修改行的比例 (0.0-1.0)
	DeleteRatio        float64       // 目标库中删除行的比例 (模拟源独有) (0.0-1.0)
	InsertExtraRatio   float64       // 目标库中新增行的比例 (模拟目标独有) (0.0-1.0)
	CleanupTarget      bool          // 在插入前清空目标表
	IncludeTypes       bool          // 是否包含 checksum_test_types 表
	IncludeComposite   bool          // 是否包含 checksum_test_composite_key 表
	IncludeEmpty       bool          // 是否包含 checksum_test_empty 表
	NULLProbability    float64       // 生成 NULL 值的概率 (0.0-1.0)
	Concurrency        int           // 插入数据的并行协程数量
	BatchTimeout       time.Duration // 新增: 批量插入的超时时间
}

var cfg Config
var logger zerolog.Logger // 全局日志记录器

// --- 模拟中国人风格的数据组件 ---
var surnames = []string{"赵", "钱", "孙", "李", "周", "吴", "郑", "王", "冯", "陈", "褚", "卫", "蒋", "沈", "韩", "杨", "朱", "秦", "尤", "许"}
var givenNames = []string{"伟", "芳", "敏", "杰", "静", "亮", "娜", "秀英", "建华", "丽", "强", "梅", "军", "桂英", "波", "兰", "青", "军", "霞", "刚"}
var cities = []string{"北京", "上海", "广州", "深圳", "杭州", "成都", "重庆", "武汉", "苏州", "西安", "南京", "天津", "郑州", "长沙", "沈阳", "青岛"}
var streetNames = []string{"人民路", "解放街", "建设大道", "中山路", "东方广场", "世纪大道", "青年路", "幸福街", "光明路", "和平街", "团结路", "工业园"}
var jobTitles = []string{"工程师", "经理", "专员", "助理", "总监", "主管", "架构师", "设计师", "分析师", "产品经理", "销售代表", "客服"}

// 中文常用字符集 (用于生成随机中文字符串)
var chineseChars = []int{
	0x4e00, 0x4e01, 0x4e03, 0x4e07, 0x4e08, 0x4e09, 0x4e0a, 0x4e0b, 0x4e0c, 0x4e0d,
	0x4e0e, 0x4e10, 0x4e11, 0x4e13, 0x4e14, 0x4e15, 0x4e16, 0x4e18, 0x4e19, 0x4e1a,
	0x4e1b, 0x4e1c, 0x4e1d, 0x4e22, 0x4e24, 0x4e25, 0x4e27, 0x4e28, 0x4e2a, 0x4e2b,
	0x4e2c, 0x4e2d, 0x4e30, 0x4e32, 0x4e34, 0x4e36, 0x4e38, 0x4e39, 0x4e3a, 0x4e3b,
	0x4e3d, 0x4e3e, 0x4e3f, 0x4e43, 0x4e45, 0x4e47, 0x4e48, 0x4e49, 0x4e4b, 0x4e4c,
	0x4e4d, 0x4e4e, 0x4e4f, 0x4e50, 0x4e52, 0x4e53, 0x4e54, 0x4e56, 0x4e58, 0x4e59,
	0x4e5c, 0x4e5d, 0x4e5e, 0x4e5f, 0x4e60, 0x4e61, 0x4e66, 0x4e69, 0x4e70, 0x4e71,
	0x4e73, 0x4e7e, 0x4e86, 0x4e88, 0x4e89, 0x4e8b, 0x4e8c, 0x4e8e, 0x4e8f, 0x4e91,
	0x4e92, 0x4e93, 0x4e94, 0x4e95, 0x4e98, 0x4e9a, 0x4e9b, 0x4e9f, 0x4ea0, 0x4ea1,
	0x4ea2, 0x4ea4, 0x4ea5, 0x4ea6, 0x4ea7, 0x4ea8, 0x4ea9, 0x4eab, 0x4eac, 0x4ead,
}

// --- 表结构定义 (包含中文注释) ---
const createTableTypesSQL = `
CREATE TABLE IF NOT EXISTS %s.%s (
    id INT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '自增主键ID',
    bool_col BOOLEAN COMMENT '布尔类型',
    tiny_col TINYINT COMMENT '有符号TinyInt (-128 to 127)',
    tiny_unsigned_col TINYINT UNSIGNED COMMENT '无符号TinyInt (0 to 255)',
    small_col SMALLINT COMMENT '有符号SmallInt (-32768 to 32767)',
    small_unsigned_col SMALLINT UNSIGNED COMMENT '无符号SmallInt (0 to 65535)',
    medium_col MEDIUMINT COMMENT '有符号MediumInt (-8388608 to 8388607)',
    medium_unsigned_col MEDIUMINT UNSIGNED COMMENT '无符号MediumInt (0 to 16777215)',
    int_col INT COMMENT '有符号Int (-2.1E9 to 2.1E9)',
    int_unsigned_col INT UNSIGNED COMMENT '无符号Int (0 to 4.2E9)',
    big_col BIGINT COMMENT '有符号BigInt (-9.2E18 to 9.2E18)',
    big_unsigned_col BIGINT UNSIGNED COMMENT '无符号BigInt (0 to 1.8E19)',
    decimal_col DECIMAL(10, 2) COMMENT '定点数Decimal (总共10位, 小数2位)',
    float_col FLOAT COMMENT '单精度浮点数',
    double_col DOUBLE COMMENT '双精度浮点数',
    char_col CHAR(10) COMMENT '定长字符串Char (长度10)',
    varchar_col VARCHAR(255) COMMENT '变长字符串Varchar (最大255)',
    tinytext_col TINYTEXT COMMENT '微小文本TinyText (最大255)',
    text_col TEXT COMMENT '普通文本Text (最大64KB)',
    mediumtext_col MEDIUMTEXT COMMENT '中等文本MediumText (最大16MB)',
    longtext_col LONGTEXT COMMENT '长文本LongText (最大4GB)',
    binary_col BINARY(16) COMMENT '定长二进制Binary (长度16)',
    varbinary_col VARBINARY(255) COMMENT '变长二进制Varbinary (最大255)',
    tinyblob_col TINYBLOB COMMENT '微小二进制TinyBlob (最大255)',
    blob_col BLOB COMMENT '普通二进制Blob (最大64KB)',
    mediumblob_col MEDIUMBLOB COMMENT '中等二进制MediumBlob (最大16MB)',
    longblob_col LONGBLOB COMMENT '长二进制LongBlob (最大4GB)',
    date_col DATE COMMENT '日期类型 (YYYY-MM-DD)',
    time_col TIME COMMENT '时间类型 (HH:MM:SS)',
    datetime_col DATETIME COMMENT '日期时间类型 (YYYY-MM-DD HH:MM:SS)',
    datetime_col_was_ts DATETIME NULL DEFAULT NULL COMMENT '模拟原Timestamp列,允许NULL (YYYY-MM-DD HH:MM:SS.ffffff)',
    year_col YEAR COMMENT '年份类型 (1901-2155)',
    enum_col ENUM('apple', 'banana', 'orange') COMMENT '枚举类型',
    set_col SET('red', 'green', 'blue') COMMENT '集合类型',
    json_col JSON COMMENT 'JSON类型',
    nullable_int INT NULL COMMENT '可空Int',
    nullable_varchar VARCHAR(50) NULL COMMENT '可空Varchar (最大50)',
    nullable_text TEXT NULL COMMENT '可空Text',
    nullable_datetime DATETIME NULL COMMENT '可空Datetime',
    nullable_blob BLOB NULL COMMENT '可空Blob',
    PRIMARY KEY (id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT '测试各种数据类型';
`

const createTableCompositeKeySQL = `
CREATE TABLE IF NOT EXISTS %s.%s (
    user_id INT UNSIGNED NOT NULL COMMENT '用户ID',
    item_id INT UNSIGNED NOT NULL COMMENT '商品ID',
    description VARCHAR(100) COMMENT '商品描述',
    quantity INT COMMENT '购买数量',
    last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '最后更新时间戳',
    PRIMARY KEY (user_id, item_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT '测试复合主键';
`

const createTableEmptySQL = `
CREATE TABLE IF NOT EXISTS %s.%s (
    id INT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '自增主键ID',
    name VARCHAR(50) COMMENT '名称',
    PRIMARY KEY (id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT '测试空表';
`

// main 函数是程序入口，负责初始化配置、连接数据库并协调数据生成过程
func main() {
	// 解析命令行参数
	flag.StringVar(&cfg.SourceDSN, "source-dsn", "root:123456@tcp(127.0.0.1:3306)/checksum_test_db?charset=utf8mb4&collation=utf8mb4_unicode_ci&parseTime=true", "源数据库连接字符串")
	flag.StringVar(&cfg.TargetDSN, "target-dsn", "root:123456@tcp(127.0.0.1:3307)/checksum_test_db?charset=utf8mb4&collation=utf8mb4_unicode_ci&parseTime=true", "目标数据库连接字符串")
	flag.StringVar(&cfg.SchemaName, "schema-name", "checksum_test_db", "数据库模式名")
	flag.StringVar(&cfg.TableTypesName, "table-types-name", "checksum_test_types", "各种类型测试表名")
	flag.StringVar(&cfg.TableCompositeName, "table-composite-name", "checksum_test_composite_key", "复合键测试表名")
	flag.StringVar(&cfg.TableEmptyName, "table-empty-name", "checksum_test_empty", "空表表名")
	flag.IntVar(&cfg.RowCount, "row-count", 10, "主表 (types/composite) 要生成的总行数")
	flag.IntVar(&cfg.BatchSize, "batch-size", 2, "批量插入的行数 (已调整，避免超过MySQL占位符限制)")
	flag.Float64Var(&cfg.ModifyRatio, "modify-ratio", 0.002, "目标库中修改行的比例 (0.0-1.0)")
	flag.Float64Var(&cfg.DeleteRatio, "delete-ratio", 0.001, "目标库中删除行的比例 (模拟源独有) (0.0-1.0)")
	flag.Float64Var(&cfg.InsertExtraRatio, "insert-extra-ratio", 0.001, "目标库中新增行的比例 (模拟目标独有) (0.0-1.0)")
	flag.BoolVar(&cfg.CleanupTarget, "cleanup-target", true, "插入前清空目标表")
	flag.BoolVar(&cfg.IncludeTypes, "include-types", true, "包含 checksum_test_types 表")
	flag.BoolVar(&cfg.IncludeComposite, "include-composite", true, "包含 checksum_test_composite_key 表")
	flag.BoolVar(&cfg.IncludeEmpty, "include-empty", true, "包含 checksum_test_empty 表")
	flag.Float64Var(&cfg.NULLProbability, "null-prob", 0.1, "生成 NULL 值的概率 (0.0-1.0)")
	flag.IntVar(&cfg.Concurrency, "concurrency", runtime.NumCPU(), "插入数据的并行协程数量 (默认为CPU核心数)")
	flag.DurationVar(&cfg.BatchTimeout, "batch-timeout", 30*time.Second, "批量插入的超时时间") // 新增超时参数

	flag.Parse()

	// 初始化随机数种子，使用当前时间确保每次运行生成不同的随机数据
	rand.Seed(time.Now().UnixNano())

	// 配置日志输出格式为非JSON格式，并简化文件调用行显示
	consoleWriter := zerolog.ConsoleWriter{
		Out:        os.Stdout,
		TimeFormat: "15:04:05",
		NoColor:    false,
	}
	// 自定义输出格式，使文件调用行更简短
	consoleWriter.FormatCaller = func(i interface{}) string {
		if i == nil {
			return ""
		}
		caller, ok := i.(string)
		if !ok {
			return ""
		}
		// 只保留文件名和行号，不显示完整路径
		parts := strings.Split(caller, "/")
		if len(parts) > 0 {
			return parts[len(parts)-1]
		}
		return caller
	}
	zerolog.SetGlobalLevel(zerolog.InfoLevel)
	logger = zerolog.New(consoleWriter).With().Timestamp().Caller().Logger()
	//logger.Level(zerolog.InfoLevel)

	// --- 连接数据库 ---
	logger.Info().Msg("正在连接源数据库...")
	sourceDB, err := sql.Open("mysql", cfg.SourceDSN)
	if err != nil {
		logger.Error().Err(err).Msg("连接源数据库失败")
		os.Exit(1)
	}
	defer sourceDB.Close()
	// 设置连接池参数以优化性能
	sourceDB.SetMaxOpenConns(cfg.Concurrency * 2) // 最大连接数设为并发数的两倍
	sourceDB.SetMaxIdleConns(cfg.Concurrency)     // 最大空闲连接数设为并发数
	sourceDB.SetConnMaxLifetime(time.Hour)        // 连接最大生命周期为1小时

	if err := sourceDB.Ping(); err != nil {
		logger.Error().Err(err).Msg("Ping 源数据库失败")
		os.Exit(1)
	}
	logger.Info().Msg("源数据库连接成功。")

	logger.Info().Msg("正在连接目标数据库...")
	targetDB, err := sql.Open("mysql", cfg.TargetDSN)
	if err != nil {
		logger.Error().Err(err).Msg("连接目标数据库失败")
		os.Exit(1)
	}
	defer targetDB.Close()
	// 设置连接池参数以优化性能
	targetDB.SetMaxOpenConns(cfg.Concurrency * 2) // 最大连接数设为并发数的两倍
	targetDB.SetMaxIdleConns(cfg.Concurrency)     // 最大空闲连接数设为并发数
	targetDB.SetConnMaxLifetime(time.Hour)        // 连接最大生命周期为1小时

	if err := targetDB.Ping(); err != nil {
		logger.Error().Err(err).Msg("Ping 目标数据库失败")
		os.Exit(1)
	}
	logger.Info().Msg("目标数据库连接成功。")

	ctx := context.Background()

	// --- 创建表 ---
	logger.Info().Msg("正在创建测试表...")
	if cfg.IncludeTypes {
		if _, err := sourceDB.ExecContext(ctx, fmt.Sprintf(createTableTypesSQL, cfg.SchemaName, cfg.TableTypesName)); err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableTypesName).Msg("创建源库表失败")
			os.Exit(1)
		}
		if _, err := targetDB.ExecContext(ctx, fmt.Sprintf(createTableTypesSQL, cfg.SchemaName, cfg.TableTypesName)); err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableTypesName).Msg("创建目标库表失败")
			os.Exit(1)
		}
		logger.Info().Str("schema", cfg.SchemaName).Str("table", cfg.TableTypesName).Msg("表已创建")
	}
	if cfg.IncludeComposite {
		if _, err := sourceDB.ExecContext(ctx, fmt.Sprintf(createTableCompositeKeySQL, cfg.SchemaName, cfg.TableCompositeName)); err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableCompositeName).Msg("创建源库表失败")
			os.Exit(1)
		}
		if _, err := targetDB.ExecContext(ctx, fmt.Sprintf(createTableCompositeKeySQL, cfg.SchemaName, cfg.TableCompositeName)); err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableCompositeName).Msg("创建目标库表失败")
			os.Exit(1)
		}
		logger.Info().Str("schema", cfg.SchemaName).Str("table", cfg.TableCompositeName).Msg("表已创建")
	}
	if cfg.IncludeEmpty {
		// 确保空表在源和目标都存在且为空
		dropEmptySQL := fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`", cfg.SchemaName, cfg.TableEmptyName)
		if _, err := sourceDB.ExecContext(ctx, dropEmptySQL); err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableEmptyName).Msg("删除源库旧空表失败")
			os.Exit(1)
		}
		if _, err := targetDB.ExecContext(ctx, dropEmptySQL); err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableEmptyName).Msg("删除目标库旧空表失败")
			os.Exit(1)
		}
		if _, err := sourceDB.ExecContext(ctx, fmt.Sprintf(createTableEmptySQL, cfg.SchemaName, cfg.TableEmptyName)); err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableEmptyName).Msg("创建源库空表失败")
			os.Exit(1)
		}
		if _, err := targetDB.ExecContext(ctx, fmt.Sprintf(createTableEmptySQL, cfg.SchemaName, cfg.TableEmptyName)); err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableEmptyName).Msg("创建目标库空表失败")
			os.Exit(1)
		}
		logger.Info().Str("schema", cfg.SchemaName).Str("table", cfg.TableEmptyName).Msg("表已创建")
	}
	logger.Info().Msg("测试表创建完成")

	// --- 清空目标表数据 ---
	logger.Info().Msg("正在清空目标表数据...")

	// 先清空源库表
	if cfg.IncludeTypes {
		_, err = sourceDB.Exec(fmt.Sprintf("TRUNCATE TABLE %s.%s", cfg.SchemaName, cfg.TableTypesName))
		if err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableTypesName).Msg("清空表失败")
			os.Exit(1)
		}
		logger.Info().Str("schema", cfg.SchemaName).Str("table", cfg.TableTypesName).Msg("表已清空")
	}

	if cfg.IncludeComposite {
		_, err = sourceDB.Exec(fmt.Sprintf("TRUNCATE TABLE %s.%s", cfg.SchemaName, cfg.TableCompositeName))
		if err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableCompositeName).Msg("清空表失败")
			os.Exit(1)
		}
		logger.Info().Str("schema", cfg.SchemaName).Str("table", cfg.TableCompositeName).Msg("表已清空")
	}

	// 再清空目标库表
	if cfg.IncludeTypes && cfg.CleanupTarget {
		_, err = targetDB.Exec(fmt.Sprintf("TRUNCATE TABLE %s.%s", cfg.SchemaName, cfg.TableTypesName))
		if err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableTypesName).Msg("清空表失败")
			os.Exit(1)
		}
		logger.Info().Str("schema", cfg.SchemaName).Str("table", cfg.TableTypesName).Msg("表已清空")
	}

	if cfg.IncludeComposite && cfg.CleanupTarget {
		_, err = targetDB.Exec(fmt.Sprintf("TRUNCATE TABLE %s.%s", cfg.SchemaName, cfg.TableCompositeName))
		if err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableCompositeName).Msg("清空表失败")
			os.Exit(1)
		}
		logger.Info().Str("schema", cfg.SchemaName).Str("table", cfg.TableCompositeName).Msg("表已清空")
	}

	logger.Info().Msg("目标表清空完成")

	// --- 生成并插入数据 ---
	logger.Info().Int("rowCount", cfg.RowCount).Msg("开始生成并插入数据")
	if cfg.IncludeTypes {
		logger.Info().Str("schema", cfg.SchemaName).Str("table", cfg.TableTypesName).Msg("正在处理表")
		if err := generateAndInsertData(ctx, sourceDB, targetDB, cfg.SchemaName, cfg.TableTypesName, cfg.RowCount, cfg.BatchSize, cfg.ModifyRatio, cfg.DeleteRatio, cfg.InsertExtraRatio, generateTypesRow, 1); err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableTypesName).Msg("生成并插入数据失败")
			os.Exit(1)
		}
		logger.Info().Str("schema", cfg.SchemaName).Str("table", cfg.TableTypesName).Msg("表数据生成完成")
	}

	if cfg.IncludeComposite {
		logger.Info().Str("schema", cfg.SchemaName).Str("table", cfg.TableCompositeName).Msg("正在处理表")
		if err := generateAndInsertData(ctx, sourceDB, targetDB, cfg.SchemaName, cfg.TableCompositeName, cfg.RowCount, cfg.BatchSize, cfg.ModifyRatio, cfg.DeleteRatio, cfg.InsertExtraRatio, generateCompositeRow, 1); err != nil {
			logger.Error().Err(err).Str("schema", cfg.SchemaName).Str("table", cfg.TableCompositeName).Msg("生成并插入数据失败")
			os.Exit(1)
		}
		logger.Info().Str("schema", cfg.SchemaName).Str("table", cfg.TableCompositeName).Msg("表数据生成完成")
	}

	if cfg.IncludeEmpty {
		logger.Info().Str("schema", cfg.SchemaName).Str("table", cfg.TableEmptyName).Msg("表被设置为保持为空")
	}

	logger.Info().Msg("所有测试数据生成完成")
	logger.Info().Strs("tables", []string{
		fmt.Sprintf("%s.%s", cfg.SchemaName, cfg.TableTypesName),
		fmt.Sprintf("%s.%s", cfg.SchemaName, cfg.TableCompositeName),
	}).Msg("您现在可以使用校验工具对表进行性能和功能测试")
}

// generateAndInsertData 生成并插入数据到源库和目标库
// 此函数协调生成器、批处理器和插入器协程
func generateAndInsertData(ctx context.Context, sourceDB, targetDB *sql.DB, schema, table string, rowCount int, batchSize int, modifyRatio, deleteRatio, insertExtraRatio float64, generateRowFunc func(int, string, int) map[string]interface{}, concurrency int) error {

	// --- 唯一组合生成逻辑，仅对 composite_key 表生效 ---
	var compositeCombos [][2]uint32
	var extraCombos [][2]uint32
	isComposite := (table == cfg.TableCompositeName)
	// itemRangePerUser := 3000
	if isComposite {
		// 用 map 保证组合绝对唯一
		comboSet := make(map[[2]uint32]struct{})
		compositeCombos = compositeCombos[:0]
		for len(compositeCombos) < rowCount {
			userID := uint32(1)
			itemID := uint32(len(compositeCombos) + 1)
			// 保证 user_id/item_id 递增且唯一
			combo := [2]uint32{userID, itemID}
			if _, exists := comboSet[combo]; !exists {
				comboSet[combo] = struct{}{}
				compositeCombos = append(compositeCombos, combo)
			}
		}
		// 检查池唯一性
		logger.Info().Int("size", len(compositeCombos)).Msg("【调试】compositeCombos 池大小")
		comboCheck := make(map[string]struct{})
		for i, c := range compositeCombos {
			key := fmt.Sprintf("%d-%d", c[0], c[1])
			if _, exists := comboCheck[key]; exists {
				logger.Error().Str("key", key).Msg("【致命错误】compositeCombos 池中有重复组合")
				os.Exit(1)
			}
			comboCheck[key] = struct{}{}
			if i < 10 {
				logger.Debug().Int("index", i).Uint32("user_id", c[0]).Uint32("item_id", c[1]).Msg("【调试】compositeCombos[%d] = (%d,%d)")
			}
		}
		// 确认表已清空
		var count int
		err := sourceDB.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM %s.%s", schema, table)).Scan(&count)
		if err != nil {
			logger.Warn().Str("schema", schema).Str("table", table).Err(err).Msg("【警告】检查表 %s.%s 是否为空失败")
		} else if count > 0 {
			logger.Error().Str("schema", schema).Str("table", table).Int("count", count).Msg("【致命错误】表 %s.%s 在插入前未清空，当前有 %d 行数据")
			os.Exit(1)
		} else {
			logger.Info().Str("schema", schema).Str("table", table).Msg("【调试】表 %s.%s 已确认为空")
		}
		// 打乱顺序
		rand.Shuffle(len(compositeCombos), func(i, j int) {
			compositeCombos[i], compositeCombos[j] = compositeCombos[j], compositeCombos[i]
		})
		// 再次检查池唯一性（shuffle后）
		comboCheck = make(map[string]struct{})
		for i, c := range compositeCombos {
			key := fmt.Sprintf("%d-%d", c[0], c[1])
			if _, exists := comboCheck[key]; exists {
				logger.Error().Str("key", key).Msg("【致命错误】shuffle后 compositeCombos 池中有重复组合")
				os.Exit(1)
			}
			comboCheck[key] = struct{}{}
			if i < 10 {
				logger.Debug().Int("index", i).Uint32("user_id", c[0]).Uint32("item_id", c[1]).Msg("【调试】shuffle后 compositeCombos[%d] = (%d,%d)")
			}
		}
	}

	// 生成 extra 组合池
	extraUserOffset := uint32(100000)
	extraCount := int(float64(rowCount) * insertExtraRatio)
	extraCombos = extraCombos[:0]
	comboSetExtra := make(map[[2]uint32]struct{})
	for len(extraCombos) < extraCount {
		userID := uint32(1 + extraUserOffset)
		itemID := uint32(len(extraCombos) + 1)
		combo := [2]uint32{userID, itemID}
		if _, exists := comboSetExtra[combo]; !exists {
			comboSetExtra[combo] = struct{}{}
			extraCombos = append(extraCombos, combo)
		}
	}
	// 检查 extra 池唯一性
	logger.Info().Int("size", len(extraCombos)).Msg("【调试】extraCombos 池大小")
	extraComboCheck := make(map[string]struct{})
	for i, c := range extraCombos {
		key := fmt.Sprintf("%d-%d", c[0], c[1])
		if _, exists := extraComboCheck[key]; exists {
			logger.Error().Str("key", key).Msg("【致命错误】extraCombos 池中有重复组合")
			os.Exit(1)
		}
		extraComboCheck[key] = struct{}{}
		if i < 5 {
			logger.Debug().Int("index", i).Uint32("user_id", c[0]).Uint32("item_id", c[1]).Msg("【调试】extraCombos[%d] = (%d,%d)")
		}
	}
	// 打乱 extra 组合顺序
	rand.Shuffle(len(extraCombos), func(i, j int) {
		extraCombos[i], extraCombos[j] = extraCombos[j], extraCombos[i]
	})

	// --- 源库阶段 ---
	logger.Info().Str("schema", schema).Str("table", table).Int("rowCount", rowCount).Msg("源库阶段: 开始处理表")
	sourceCtx, sourceCancel := context.WithCancel(ctx)
	defer sourceCancel()

	sourceRowChan := make(chan map[string]interface{}, batchSize*2)
	sourceBatchChan := make(chan []map[string]interface{}, concurrency*2)
	errChan := make(chan error, concurrency+3)

	var wg sync.WaitGroup

	// 1. 源库生成器
	wg.Add(1)
	go func() {
		defer wg.Done()
		defer close(sourceRowChan)
		logger.Info().Msg("源库生成器启动...")
		if isComposite {
			// 检查源库生成器是否被多次启动
			logger.Info().Int("rowCount", rowCount).Msg("【调试】源库生成器启动，将生成数据")
			// 记录已用组合，确保不重复
			usedCombos := make(map[string]struct{})
			for i := 0; i < rowCount; i++ {
				userID := compositeCombos[i][0]
				itemID := compositeCombos[i][1]
				key := fmt.Sprintf("%d-%d", userID, itemID)
				if _, exists := usedCombos[key]; exists {
					// Skip this combination and try another one
					continue
				}
				usedCombos[key] = struct{}{}
				row := generateCompositeRowWithKey(userID, itemID, "source", rowCount)
				sourceRowChan <- row
				if i < 10 || (i > 0 && i%100000 == 0) {
					logger.Debug().Int("index", i).Uint32("user_id", userID).Uint32("item_id", itemID).Msg("【调试】源库生成行")
				}
			}
			logger.Info().Int("rowCount", len(usedCombos)).Msg("【调试】源库生成器完成，所有组合均唯一")
		} else {
			for i := 0; i < rowCount; i++ {
				select {
				case <-sourceCtx.Done():
					logger.Info().Msg("源库生成器收到取消信号，退出")
					return
				default:
					// continue
				}
				sourceRowChan <- generateRowFunc(i, "source", rowCount)
				if i > 0 && i%100000 == 0 {
					logger.Info().Int("rowCount", i).Msg("源库生成器已生成行数")
				}
			}
			logger.Info().Int("rowCount", rowCount).Msg("源库生成器完成")
		}
	}()

	// 2. 源库批处理器
	wg.Add(1)
	go func() {
		defer wg.Done()
		defer close(sourceBatchChan)
		logger.Info().Msg("源库批处理器启动...")
		// 记录批处理器已处理的组合，用于检测重复数据
		batchProcessedCombos := make(map[string]struct{})
		batch := make([]map[string]interface{}, 0, batchSize) // 预分配容量，减少内存分配
		count := 0                                            // 记录已处理的总行数
		for row := range sourceRowChan {
			select {
			case <-sourceCtx.Done():
				logger.Info().Msg("源库批处理器收到取消信号，退出")
				return
			default:
				// 继续执行
			}
			batch = append(batch, row)
			if isComposite {
				// 对于复合键表，需要检查是否有重复组合
				key := fmt.Sprintf("%d-%d", row["user_id"], row["item_id"])
				if _, exists := batchProcessedCombos[key]; exists {
					logger.Error().Str("key", key).Msg("【致命错误】批处理器收到重复组合")
					os.Exit(1)
				}
				batchProcessedCombos[key] = struct{}{}
				if count < 10 {
					// 记录前10行数据用于调试
					logger.Debug().Int("index", count).Interface("user_id", row["user_id"]).Interface("item_id", row["item_id"]).Msg("【调试】批处理器处理行")
				}
			}
			if len(batch) == batchSize {
				// 批次已满，发送到批次通道
				count += len(batch)
				logger.Info().Int("worker", 0).Int("batchSize", batchSize).Int("count", count).Msg("源库批处理器尝试发送满批次")
				select {
				case sourceBatchChan <- batch:
					logger.Info().Int("worker", 0).Msg("源库批处理器成功发送满批次")
					batch = make([]map[string]interface{}, 0, batchSize) // 创建新批次
				case <-sourceCtx.Done():
					logger.Info().Int("worker", 0).Msg("源库批处理器发送批量时收到取消信号，退出")
					return
				}
			}
		}
		// 处理剩余不足一个批次的数据
		if len(batch) > 0 {
			count += len(batch)
			logger.Info().Int("worker", 0).Int("batchSize", len(batch)).Int("count", count).Msg("源库批处理器尝试发送最后一个批次")
			select {
			case sourceBatchChan <- batch:
				logger.Info().Int("worker", 0).Msg("源库批处理器成功发送最后一个批次")
			case <-sourceCtx.Done():
				logger.Info().Int("worker", 0).Msg("源库批处理器发送最后批次时收到取消信号，退出")
				return
			}
		}
		logger.Info().Int("worker", 0).Int("count", count).Msg("源库批处理器已处理完所有单行数据")
	}()
	// 3. 源库插入器池（修正：插入器数量不超过批次数，且每个批次只被一个插入器消费）
	batchCount := (rowCount + batchSize - 1) / batchSize // 计算总批次数，向上取整
	workerCount := batchCount
	if workerCount > concurrency {
		workerCount = concurrency // 插入器数量不超过配置的并发数
	}
	logger.Info().Int("workerCount", workerCount).Int("batchCount", batchCount).Msg("源库插入器池启动")

	// 启动多个插入器协程并行处理批次数据
	for i := 0; i < workerCount; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			logger.Info().Int("worker", workerID).Msg("源库插入器启动")
			batchesProcessed := 0 // 记录该插入器处理的批次数

			// 循环从批次通道获取数据并插入数据库
			for batch := range sourceBatchChan {
				select {
				case <-sourceCtx.Done():
					logger.Info().Int("worker", workerID).Msg("源库插入器收到取消信号，退出")
					return
				default:
					// 继续执行
				}

				// 检查复合键表的批次是否有重复组合
				if isComposite {
					if err := checkInsertion(batch); err != nil {
						logger.Error().Int("worker", workerID).Err(err).Msg("【致命错误】源库插入器发现批次中有重复组合")
						errChan <- err
						sourceCancel() // 取消所有源库协程
						return
					}
				}

				// 设置插入超时上下文
				insertCtx, insertCancel := context.WithTimeout(sourceCtx, cfg.BatchTimeout)
				// 执行批量插入操作
				startTime := time.Now()
				err := insertRowsInBatches(insertCtx, sourceDB, schema, table, batch, batchSize)
				insertCancel() // 释放超时上下文

				if err != nil {
					// 插入失败，记录错误并取消所有协程
					logger.Error().Int("worker", workerID).Err(err).Msg("源库插入失败")
					errChan <- err
					sourceCancel()
					return
				}

				// 插入成功，记录性能指标
				elapsed := time.Since(startTime)
				batchesProcessed++
				rowsInserted := len(batch)

				// 记录插入性能日志
				logger.Info().
					Int("worker", workerID).
					Int("batchSize", rowsInserted).
					Int("batchesProcessed", batchesProcessed).
					Float64("elapsed", elapsed.Seconds()).
					Float64("rowsPerSecond", float64(rowsInserted)/elapsed.Seconds()).
					Msg("源库插入成功")
			}

			logger.Info().Int("worker", workerID).Int("batchesProcessed", batchesProcessed).Msg("源库插入器完成")
		}(i)
	}

	go func() {
		wg.Wait()
		close(errChan)
		logger.Info().Msg("源库阶段所有协程已完成或退出，错误通道已关闭")
	}()

	if err := <-errChan; err != nil {
		return fmt.Errorf("源库数据生成或插入过程中发生错误: %w", err)
	}
	logger.Info().Msg("源库阶段完成")

	// --- 目标库阶段 ---
	logger.Info().Str("schema", schema).Str("table", table).Int("rowCount", rowCount).Msg("目标库阶段: 开始处理表")
	targetCtx, targetCancel := context.WithCancel(ctx)
	defer targetCancel()

	targetRowChan := make(chan map[string]interface{}, batchSize*2)
	targetBatchChan := make(chan []map[string]interface{}, concurrency*2)
	targetErrChan := make(chan error, concurrency+3)

	var targetWg sync.WaitGroup

	// 1. 目标库生成器
	targetWg.Add(1)
	go func() {
		defer targetWg.Done()
		defer close(targetRowChan)
		logger.Info().Msg("目标库生成器启动...")

		if isComposite {
			totalTargetRowsToGenerate := rowCount + int(float64(rowCount)*insertExtraRatio)
			numModified := 0
			numDeleted := 0
			numExtraInserted := 0
			modCombos := make([][2]uint32, len(compositeCombos))
			copy(modCombos, compositeCombos)
			rand.Shuffle(len(modCombos), func(i, j int) {
				modCombos[i], modCombos[j] = modCombos[j], modCombos[i]
			})
			extraUsed := 0
			// Add a map to track all used combinations in the target database
			targetUsedCombos := make(map[string]struct{})
			for i := 0; i < totalTargetRowsToGenerate; i++ {
				select {
				case <-targetCtx.Done():
					logger.Info().Msg("目标库生成器收到取消信号，退出")
					return
				default:
					// continue
				}
				// 决定当前行是否为源库中被删除的行
				isDeleted := false
				if i < rowCount && rand.Float64() < deleteRatio {
					numDeleted++
					isDeleted = true
				}
				if !isDeleted {
					var generatedRow map[string]interface{}
					if i < rowCount && rand.Float64() < modifyRatio {
						userID := modCombos[i][0]
						itemID := modCombos[i][1]
						// Check if this combination has already been used
						key := fmt.Sprintf("%d-%d", userID, itemID)
						if _, exists := targetUsedCombos[key]; exists {
							// Skip this combination and try another one
							continue
						}
						targetUsedCombos[key] = struct{}{}
						generatedRow = generateCompositeRowWithKey(userID, itemID, "modified", rowCount)
						numModified++
					} else if i < rowCount {
						userID := compositeCombos[i][0]
						itemID := compositeCombos[i][1]
						// Check if this combination has already been used
						key := fmt.Sprintf("%d-%d", userID, itemID)
						if _, exists := targetUsedCombos[key]; exists {
							// Skip this combination and try another one
							continue
						}
						targetUsedCombos[key] = struct{}{}
						generatedRow = generateCompositeRowWithKey(userID, itemID, "source", rowCount)
					} else {
						if extraUsed < len(extraCombos) {
							userID := extraCombos[extraUsed][0]
							itemID := extraCombos[extraUsed][1]
							generatedRow = generateCompositeRowWithKey(userID, itemID, "extra", rowCount)
							extraUsed++
							numExtraInserted++
						}
					}
					if generatedRow != nil {
						targetRowChan <- generatedRow
					}
				}
				if i > 0 && i%100000 == 0 {
					logger.Info().Int("rowCount", i).Msg("目标库生成器已生成行数")
				}
			}
			logger.Info().
				Int("totalTargetRowsToGenerate", totalTargetRowsToGenerate).
				Int("rowCount", rowCount).
				Int("numDeleted", numDeleted).
				Int("numModified", numModified).
				Int("numExtraInserted", numExtraInserted).
				Int("actualRows", totalTargetRowsToGenerate-numDeleted).
				Msg("目标库生成器完成")
		} else {
			// 非 composite 表逻辑保持不变
			totalTargetRowsToGenerate := rowCount + int(float64(rowCount)*insertExtraRatio)
			numModified := 0
			numDeleted := 0
			numExtraInserted := 0
			for i := 0; i < totalTargetRowsToGenerate; i++ {
				select {
				case <-targetCtx.Done():
					logger.Info().Msg("目标库生成器收到取消信号，退出")
					return
				default:
					// continue
				}
				isDeleted := false
				if i < rowCount && rand.Float64() < deleteRatio {
					numDeleted++
					isDeleted = true
				}
				if !isDeleted {
					var generatedRow map[string]interface{}
					if i < rowCount && rand.Float64() < modifyRatio {
						generatedRow = generateRowFunc(i, "modified", rowCount)
						numModified++
					} else if i < rowCount {
						generatedRow = generateRowFunc(i, "source", rowCount)
					} else {
						generatedRow = generateRowFunc(i, "extra", rowCount)
						numExtraInserted++
					}
					if generatedRow != nil {
						targetRowChan <- generatedRow
					}
				}
				if i > 0 && i%100000 == 0 {
					logger.Info().Int("rowCount", i).Msg("目标库生成器已生成行数")
				}
			}
			logger.Info().
				Int("totalTargetRowsToGenerate", totalTargetRowsToGenerate).
				Int("rowCount", rowCount).
				Int("numDeleted", numDeleted).
				Int("numModified", numModified).
				Int("numExtraInserted", numExtraInserted).
				Int("actualRows", totalTargetRowsToGenerate-numDeleted).
				Msg("目标库生成器完成")
		}
	}()
	// 2. 目标库批处理器
	targetWg.Add(1)
	go func() {
		defer targetWg.Done()
		defer close(targetBatchChan)
		logger.Info().Msg("目标库批处理器启动...")
		// 以下为批处理器部分添加的详细中文注释
		// 记录批处理器已处理的组合，用于检测重复数据
		batchProcessedCombos := make(map[string]struct{})
		batch := make([]map[string]interface{}, 0, batchSize) // 预分配容量，减少内存分配
		count := 0                                            // 记录已处理的总行数
		for row := range targetRowChan {
			select {
			case <-targetCtx.Done():
				logger.Info().Msg("目标库批处理器收到取消信号，退出")
				return
			default:
				// 继续执行
			}
			batch = append(batch, row)
			if isComposite {
				// 对于复合键表，需要检查是否有重复组合
				key := fmt.Sprintf("%d-%d", row["user_id"], row["item_id"])
				if _, exists := batchProcessedCombos[key]; exists {
					logger.Error().Str("key", key).Msg("【致命错误】批处理器收到重复组合")
					os.Exit(1)
				}
				batchProcessedCombos[key] = struct{}{}
				if count < 10 {
					// 记录前10行数据用于调试
					logger.Debug().Int("index", count).Interface("user_id", row["user_id"]).Interface("item_id", row["item_id"]).Msg("【调试】批处理器处理行")
				}
			}
			if len(batch) == batchSize {
				// 批次已满，发送到批次通道
				count += len(batch)
				logger.Info().Int("worker", 0).Int("batchSize", batchSize).Int("count", count).Msg("目标库批处理器尝试发送满批次")
				select {
				case targetBatchChan <- batch:
					logger.Info().Int("worker", 0).Msg("目标库批处理器成功发送满批次")
					batch = make([]map[string]interface{}, 0, batchSize) // 创建新批次
				case <-targetCtx.Done():
					logger.Info().Int("worker", 0).Msg("目标库批处理器发送批量时收到取消信号，退出")
					return
				}
			}
		}
		// 处理剩余不足一个批次的数据
		if len(batch) > 0 {
			count += len(batch)
			logger.Info().Int("worker", 0).Int("batchSize", len(batch)).Int("count", count).Msg("目标库批处理器尝试发送最后一个批次")
			select {
			case targetBatchChan <- batch:
				logger.Info().Int("worker", 0).Msg("目标库批处理器成功发送最后一个批次")
			case <-targetCtx.Done():
				logger.Info().Int("worker", 0).Msg("目标库批处理器发送最后批次时收到取消信号，退出")
				return
			}
		}
		logger.Info().Int("worker", 0).Int("count", count).Msg("目标库批处理器已处理完所有单行数据")
	}()
	// 3. 目标库插入器池
	targetWorkerCount := batchCount
	if targetWorkerCount > 1 {
		targetWorkerCount = 1 // 强制只启动1个插入器，彻底避免并发主键冲突
	}
	logger.Info().Int("targetWorkerCount", targetWorkerCount).Msg("启动目标库插入器协程")
	for i := 0; i < targetWorkerCount; i++ {
		targetWg.Add(1)
		go func(workerID int) {
			defer targetWg.Done()
			stageName := "目标库"
			logger.Info().Str("stage", stageName).Int("worker", workerID).Msg("插入器启动")
			for batch := range targetBatchChan {
				select {
				case <-targetCtx.Done():
					logger.Info().Str("stage", stageName).Int("worker", workerID).Msg("插入器收到取消信号，退出")
					return
				default:
					// 继续执行
				}
				batchCtx, cancelBatch := context.WithTimeout(targetCtx, cfg.BatchTimeout)
				err := insertRowsInBatches(batchCtx, targetDB, schema, table, batch, batchSize)
				cancelBatch()
				if err != nil {
					logger.Error().Str("stage", stageName).Int("worker", workerID).Int("batchSize", len(batch)).Err(err).Msg("插入器插入批量失败")
					select {
					case targetErrChan <- fmt.Errorf("%s 插入失败: %w", stageName, err):
					case <-targetCtx.Done():
					}
					targetCancel()
					return
				}
			}
			logger.Info().Str("stage", stageName).Int("worker", workerID).Msg("插入器完成")
		}(i)
	}

	go func() {
		targetWg.Wait()
		close(targetErrChan)
		logger.Info().Msg("目标库阶段所有协程已完成或退出，错误通道已关闭")
	}()

	if err := <-targetErrChan; err != nil {
		return fmt.Errorf("目标库数据生成或插入过程中发生错误: %w", err)
	}
	logger.Info().Msg("目标库阶段完成")

	// --- 最终统计 ---
	// 获取源库和目标库的行数
	var sourceCount, targetCount int
	sourceCountQuery := fmt.Sprintf("SELECT COUNT(*) FROM %s.%s", schema, table)
	targetCountQuery := fmt.Sprintf("SELECT COUNT(*) FROM %s.%s", schema, table)
	err := sourceDB.QueryRow(sourceCountQuery).Scan(&sourceCount)
	if err != nil {
		logger.Error().Err(err).Str("schema", schema).Str("table", table).Msg("获取源库表行数失败")
		return err
	}
	err = targetDB.QueryRow(targetCountQuery).Scan(&targetCount)
	if err != nil {
		logger.Error().Err(err).Str("schema", schema).Str("table", table).Msg("获取目标库表行数失败")
		return err
	}
	logger.Info().
		Str("schema", schema).
		Str("table", table).
		Int("sourceCount", sourceCount).
		Int("targetCount", targetCount).
		Int("diff", targetCount-sourceCount).
		Msg("表数据生成完成，行数统计")

	return nil
}

// checkInsertion 检查批次中是否有重复组合
// 用于确保复合键表中的数据唯一性，防止主键冲突
func checkInsertion(batch []map[string]interface{}) error {
	// 使用map记录已存在的组合
	seen := make(map[string]struct{})
	for _, row := range batch {
		// 提取user_id和item_id
		userID, ok1 := row["user_id"]
		itemID, ok2 := row["item_id"]
		if !ok1 || !ok2 {
			// 如果行中没有这两个字段，则跳过检查
			continue
		}
		// 生成唯一键
		key := fmt.Sprintf("%v-%v", userID, itemID)
		if _, exists := seen[key]; exists {
			// 发现重复组合，返回错误
			return fmt.Errorf("批次中发现重复组合: %s", key)
		}
		seen[key] = struct{}{}
	}
	return nil
}

// insertRowsInBatches 批量插入行到指定表
// 使用参数化查询提高插入效率并防止SQL注入
func insertRowsInBatches(ctx context.Context, db *sql.DB, schema, table string, rows []map[string]interface{}, batchSize int) error {
	if len(rows) == 0 {
		// 没有数据需要插入，直接返回
		return nil
	}

	// 获取第一行数据的所有列名
	var columnNames []string
	for colName := range rows[0] {
		columnNames = append(columnNames, colName)
	}

	// 对列名进行排序，确保每次生成的SQL语句列顺序一致
	sort.Strings(columnNames)

	// 构建INSERT语句
	var sb strings.Builder
	sb.WriteString(fmt.Sprintf("INSERT INTO %s.%s (", schema, table))

	// 添加列名部分
	for i, colName := range columnNames {
		if i > 0 {
			sb.WriteString(", ")
		}
		sb.WriteString(fmt.Sprintf("`%s`", colName))
	}
	sb.WriteString(") VALUES ")

	// 为每行数据添加占位符
	placeholders := make([]string, len(rows))
	for i := range rows {
		var placeholderSb strings.Builder
		placeholderSb.WriteString("(")
		for j := range columnNames {
			if j > 0 {
				placeholderSb.WriteString(", ")
			}
			placeholderSb.WriteString("?")
		}
		placeholderSb.WriteString(")")
		placeholders[i] = placeholderSb.String()
	}
	sb.WriteString(strings.Join(placeholders, ", "))

	// 准备参数值数组
	params := make([]interface{}, 0, len(rows)*len(columnNames))
	for _, row := range rows {
		for _, colName := range columnNames {
			params = append(params, row[colName])
		}
	}

	// 执行SQL语句
	_, err := db.ExecContext(ctx, sb.String(), params...)
	return err
}

// generateCompositeRow 生成复合键表的单行数据
// index: 行索引，generationType: 生成类型(source/target)，totalBaseRowCount: 总基础行数
func generateCompositeRow(index int, generationType string, totalBaseRowCount int) map[string]interface{} {
	// 根据索引计算user_id和item_id
	userID := uint32(1)
	itemID := uint32(index + 1)
	// 调用带键的生成函数
	return generateCompositeRowWithKey(userID, itemID, generationType, totalBaseRowCount)
}

// generateCompositeRowWithKey 使用指定的 user_id 和 item_id 生成复合键表的单行数据
// 允许精确控制生成的数据主键
func generateCompositeRowWithKey(userID, itemID uint32, generationType string, totalBaseRowCount int) map[string]interface{} {
	// 创建数据行
	row := make(map[string]interface{})

	// 设置主键字段
	row["user_id"] = userID
	row["item_id"] = itemID

	// 生成随机描述
	row["description"] = generateChineseString(10) + " - " + generationType

	// 随机生成购买数量(1-100)
	row["quantity"] = rand.Intn(100) + 1

	// 随机生成最后更新时间(过去30天内)
	row["last_updated"] = time.Now().AddDate(0, 0, -rand.Intn(30))

	return row
}

// generateTypesRow 生成各种类型测试表的单行数据
// 包含各种MySQL数据类型的示例值
func generateTypesRow(index int, generationType string, totalBaseRowCount int) map[string]interface{} {
	// 创建数据行，包含各种MySQL数据类型
	row := make(map[string]interface{})

	// 设置自增主键(如果是新插入的行)
	if generationType == "source" || generationType == "target_extra" || generationType == "modified" {
		// 源库数据或目标库额外数据需要设置ID
		if generationType == "target_extra" {
			// 目标库额外数据的ID需要偏移，避免与源库数据冲突
			row["id"] = uint32(index + totalBaseRowCount + 1)
		} else if generationType == "modified" {
			// 对于修改的行，使用一个大偏移量，确保唯一性
			row["id"] = uint32(index + 2*totalBaseRowCount + 1)
		} else {
			row["id"] = uint32(index + 1)
		}
	}

	// 生成布尔类型数据
	row["bool_col"] = rand.Intn(2) == 1

	// 生成整数类型数据
	row["tiny_col"] = int8(rand.Intn(256) - 128)
	row["tiny_unsigned_col"] = uint8(rand.Intn(256))
	row["small_col"] = int16(rand.Intn(65536) - 32768)
	row["small_unsigned_col"] = uint16(rand.Intn(65536))
	row["medium_col"] = int32(rand.Intn(16777216) - 8388608)
	row["medium_unsigned_col"] = uint32(rand.Intn(16777216))
	row["int_col"] = int32(rand.Intn(2147483647) - 1073741824)
	row["int_unsigned_col"] = uint32(rand.Intn(2147483647))
	row["big_col"] = int64(rand.Int63() - (1 << 62))
	row["big_unsigned_col"] = uint64(rand.Int63())

	// 生成浮点类型数据
	row["float_col"] = float32(rand.Float64() * 1000.0)
	row["double_col"] = rand.Float64() * 1000000.0
	row["decimal_col"] = fmt.Sprintf("%.2f", rand.Float64()*10000.0)

	// 生成字符串类型数据
	row["char_col"] = generateChineseString(5)
	row["varchar_col"] = generateChineseString(10)
	row["tinytext_col"] = generateChineseString(20)
	row["text_col"] = generateChineseString(50)
	row["mediumtext_col"] = generateChineseString(100)
	row["longtext_col"] = generateChineseString(200)

	// 生成二进制类型数据
	row["binary_col"] = generateRandomBytes(16)
	row["varbinary_col"] = generateRandomBytes(32)
	row["tinyblob_col"] = generateRandomBytes(64)
	row["blob_col"] = generateRandomBytes(128)
	row["mediumblob_col"] = generateRandomBytes(256)
	row["longblob_col"] = generateRandomBytes(512)

	// 生成日期时间类型数据
	now := time.Now()
	row["date_col"] = now.Format("2006-01-02")
	row["time_col"] = now.Format("15:04:05")
	row["datetime_col"] = now
	row["datetime_col_was_ts"] = now
	row["year_col"] = fmt.Sprintf("%d", 1901+rand.Intn(255))

	// 生成枚举和集合类型数据
	enumValues := []string{"apple", "banana", "orange"}
	row["enum_col"] = enumValues[rand.Intn(len(enumValues))]

	setValues := []string{"red", "green", "blue"}
	numSetValues := rand.Intn(len(setValues)) + 1
	selectedSetValues := make([]string, 0, numSetValues)
	for i := 0; i < numSetValues; i++ {
		selectedSetValues = append(selectedSetValues, setValues[i])
	}
	row["set_col"] = strings.Join(selectedSetValues, ",")

	// 生成JSON类型数据
	jsonObj := map[string]interface{}{
		"id":      rand.Intn(1000),
		"name":    generateChineseString(5),
		"tags":    []string{"tag1", "tag2", "tag3"}[0 : rand.Intn(3)+1],
		"score":   rand.Float64() * 100.0,
		"created": time.Now().Unix(),
	}
	jsonBytes, _ := json.Marshal(jsonObj)
	row["json_col"] = string(jsonBytes)

	// 生成可空字段数据，有一定概率为NULL
	if rand.Float64() >= cfg.NULLProbability {
		row["nullable_int"] = int32(rand.Intn(1000))
	} else {
		row["nullable_int"] = nil
	}

	if rand.Float64() >= cfg.NULLProbability {
		row["nullable_text"] = generateChineseString(15)
	} else {
		row["nullable_text"] = nil
	}

	if rand.Float64() >= cfg.NULLProbability {
		row["nullable_datetime"] = time.Now().AddDate(0, 0, -rand.Intn(365))
	} else {
		row["nullable_datetime"] = nil
	}

	if rand.Float64() >= cfg.NULLProbability {
		row["nullable_blob"] = generateRandomBytes(32)
	} else {
		row["nullable_blob"] = nil
	}

	return row
}

// generateChineseString 生成指定长度的中文字符串
// 从预定义的中文字符集中随机选择字符
func generateChineseString(length int) string {
	result := make([]rune, length)
	for i := range result {
		// 从中文字符集中随机选择一个字符
		result[i] = rune(chineseChars[rand.Intn(len(chineseChars))])
	}
	return string(result)
}

// generateRandomBytes 生成指定长度的随机字节数组
// 用于填充二进制类型的列
func generateRandomBytes(length int) []byte {
	result := make([]byte, length)
	rand.Read(result) // 使用加密安全的随机数生成器填充字节数组
	return result
}

// generateValueForColumn 为特定列生成值
// 根据列名和示例值类型生成适合该列的随机值
func generateValueForColumn(colName string, sampleValue interface{}) interface{} {
	// 如果示例值为 nil 或随机值小于配置的NULL概率，则返回 nil
	if sampleValue == nil || rand.Float64() < cfg.NULLProbability {
		return nil
	}

	// 根据列名特殊处理某些列
	switch colName {
	case "id":
		// ID 列通常是主键，应该保持不变
		return sampleValue
	case "user_id", "item_id":
		// 复合主键的组成部分，应该保持不变
		return sampleValue
	}

	// 根据示例值类型生成新的随机值
	switch v := sampleValue.(type) {
	case bool:
		// 布尔类型随机生成 true 或 false
		return rand.Intn(2) == 1
	case int8:
		// 有符号8位整数，范围 -128 到 127
		return int8(rand.Intn(256) - 128)
	case uint8:
		// 无符号8位整数，范围 0 到 255
		return uint8(rand.Intn(256))
	case int16:
		// 有符号16位整数，范围 -32768 到 32767
		return int16(rand.Intn(65536) - 32768)
	case uint16:
		// 无符号16位整数，范围 0 到 65535
		return uint16(rand.Intn(65536))
	case int32:
		// 有符号32位整数，范围约 -21亿 到 21亿
		return int32(rand.Intn(2147483647) - 1073741824)
	case uint32:
		// 无符号32位整数，范围 0 到 42亿多
		return uint32(rand.Intn(2147483647))
	case int64:
		// 有符号64位整数，范围非常大
		return int64(rand.Int63() - (1 << 62))
	case uint64:
		// 无符号64位整数，范围 0 到非常大
		return uint64(rand.Int63())
	case int:
		// 根据系统可能是32位或64位
		return rand.Intn(1000000) - 500000
	case float32:
		// 32位浮点数
		return float32(rand.Float64() * 1000.0)
	case float64:
		// 64位浮点数
		return rand.Float64() * 1000000.0
	case string:
		// 字符串类型需要根据列名进行特殊处理
		if colName == "enum_col" {
			// 枚举类型从预定义值中选择
			enumValues := []string{"apple", "banana", "orange"}
			return enumValues[rand.Intn(len(enumValues))]
		} else if colName == "set_col" {
			// 集合类型选择1到n个值组合
			setValues := []string{"red", "green", "blue"}
			numSetValues := rand.Intn(len(setValues)) + 1
			selectedSetValues := make([]string, 0, numSetValues)
			for i := 0; i < numSetValues; i++ {
				selectedSetValues = append(selectedSetValues, setValues[i])
			}
			return strings.Join(selectedSetValues, ",")
		} else if colName == "json_col" {
			// JSON列生成一个包含多个字段的JSON对象
			jsonObj := map[string]interface{}{
				"id":      rand.Intn(1000),
				"name":    generateChineseString(5),
				"tags":    []string{"tag1", "tag2", "tag3"}[0 : rand.Intn(3)+1],
				"score":   rand.Float64() * 100.0,
				"created": time.Now().Unix(),
			}
			jsonBytes, err := json.Marshal(jsonObj)
			if err != nil {
				logger.Error().Err(err).Msg("JSON 编码失败")
				return v
			}
			return string(jsonBytes)
		} else if strings.HasSuffix(colName, "text_col") || strings.HasSuffix(colName, "varchar_col") || strings.HasSuffix(colName, "char_col") {
			// 文本类型列生成中文字符串
			length := len(v)
			if length > 20 {
				length = 20 // 限制长度以避免生成过长的字符串
			}
			return "修改_" + generateChineseString(length)
		} else if colName == "date_col" {
			// 日期格式 YYYY-MM-DD
			year := 2000 + rand.Intn(23) // 2000-2022
			month := rand.Intn(12) + 1   // 1-12
			day := rand.Intn(28) + 1     // 1-28 (避免月份天数问题)
			return fmt.Sprintf("%04d-%02d-%02d", year, month, day)
		} else if colName == "time_col" {
			// 时间格式 HH:MM:SS
			hour := rand.Intn(24)   // 0-23
			minute := rand.Intn(60) // 0-59
			second := rand.Intn(60) // 0-59
			return fmt.Sprintf("%02d:%02d:%02d", hour, minute, second)
		} else if colName == "year_col" {
			// 年份范围 1901-2155
			return fmt.Sprintf("%d", 1901+rand.Intn(255))
		} else {
			// 其他字符串类型
			return "修改_" + v
		}
	case []byte:
		// 二进制数据生成新的随机字节
		result := make([]byte, len(v))
		rand.Read(result)
		return result
	case time.Time:
		// 时间类型在原时间基础上随机增减
		offset := rand.Intn(60*60*24*30) - 60*60*24*15 // 随机±15天
		return v.Add(time.Duration(offset) * time.Second)
	default:
		// 未知类型保持原值不变
		logger.Warn().Str("colName", colName).Str("type", fmt.Sprintf("%T", v)).Msg("未知类型的列，保持原值不变")
		return v
	}
}
